// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once

#include <experimental/filesystem>
#include <memory>
#include <string>
#include <vector>

#include "coin2/base/health.h"
#include "coin2/exchange/base/executor/archive_executor.h"
#include "coin2/exchange/base/executor/archive_system.h"
#include "coin2/exchange/base/executor/async_executor.h"
#include "coin2/exchange/base/executor/live_executor.h"
#include "coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/exchange/base/feed/distributor.h"
#include "coin2/exchange/base/feed/subscriber.h"
#include "coin2/exchange/base/symbology/product.h"
#include "presto/quant/math/moving_average.h"

namespace coin2::exchange::base::feed {

using ::coin2::exchange::base::executor::ArchiveMixin;

namespace impl {
using coin2::exchange::base::symbology::IProduct;
class IFeedSubsystem {
 public:
  explicit IFeedSubsystem(const std::string& system_name, const MarketExchangeApi& mea)
      : system_name_(system_name), mea_(mea) {}
  virtual ~IFeedSubsystem() = default;

  const std::string& system_name() const { return system_name_; }
  const MarketExchangeApi& mea() const { return mea_; }

  virtual std::vector<IProduct*> GetProducts() const = 0;
  virtual bool BanProduct(const IProduct* product) {
    return false;
  }

  virtual bool BanProduct(const std::vector<const IProduct*>& products) {
    return false;
  }
  virtual bool BanProduct(const std::vector<IProduct*>& products) {
    return false;
  }

  virtual void DebugPrint() const {}
  virtual void GetHealthReport(int indent, std::string* report) const {
    // [HealthReport]
    //     [Feed]
    //         - Huobi (OK)
    //         - Okex (STARTING, uptime=23h)
    //               a1 (..)

    for (int i = 0; i < indent * 4; i++) {
      *report += ' ';
    }
    *report += "- ";
    *report += system_name();
    *report += "(?)\n";
  }

 protected:
  const std::string system_name_;
  const MarketExchangeApi mea_;

  Health health_{};
};

}  // namespace impl

using impl::IFeedSubsystem;

// A feed subsystem for a single exchange
// hence fastfeed or archive does not use this class
// TODO(donggu): rename to ExchangeFeedSubsystem
class BaseLiveFeedSubsystem : public IFeedSubsystem, public IAsyncActor {
 public:
  BaseLiveFeedSubsystem(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      executor::LiveAsyncExecutor* live_executor)
      : IFeedSubsystem(system_name, mea), live_executor_(live_executor) {
    live_executor_->RegisterSubsystem(this);
  }

 protected:
  executor::LiveAsyncExecutor* live_executor_;
};

class FeedArchiveMixin : public ArchiveMixin {
 public:
  explicit FeedArchiveMixin(executor::ArchiveSyncExecutor* archive_executor)
      : ArchiveMixin(archive_executor) {}

  void AddCallback(IFeedSubscriber* callback) { callbacks_.push_back(callback); }
  void PrependCallback(IFeedSubscriber* callback) {
    callbacks_.insert(callbacks_.begin(), callback);
  }

  void CheckCongestion(
      int64_t feed_ts,
      int64_t wall_ts_elapsed) {
    if (!first_feed_ts_.has_value()) first_feed_ts_ = feed_ts;
    elapsed_tss_.Update(feed_ts, wall_ts_elapsed);
    cum_elapsed_ns_ += wall_ts_elapsed;
    if (!elapsed_tss_.empty()) {
      double elapsed_ns = elapsed_tss_.sum();
      double bucket_ns = kCongestionWindowNs;
      if (elapsed_ns > bucket_ns * kCongestionThreshold) {
        elapsed_tss_.clear();
        congested_window_.Update(feed_ts, 1);
        if (congested_window_.sum() >= 20) {
          CHECK(false)
              << "Congestion ratio: "
              << elapsed_ns / bucket_ns
              << " exceeds threshold of: " << kCongestionThreshold << ", 20 times";
        }
        LOG(ERROR)
            << "Congestion issue happend, ratio: "
            << elapsed_ns / bucket_ns << " = " << elapsed_ns << " / " << bucket_ns;
      } else {
        double tot_bucket_ns = std::max<double>(bucket_ns, feed_ts - *first_feed_ts_);
        SPDLOG_INFO_EVERY_NS(600'000'000'000L, feed_ts,
            "Congestion ratio: {} = {} / {}, Cumulative: {} = {} / {}",
            elapsed_ns / bucket_ns, elapsed_ns, bucket_ns,
            cum_elapsed_ns_ / tot_bucket_ns, cum_elapsed_ns_, tot_bucket_ns);
      }
    }
  }

  void onBookFeed(const FeedUpdate& upd) {
    double pre_wallts = GetCurrentTimestamp();
    for (auto* callback : callbacks_) {
      callback->onBookFeed(upd);
    }
    CheckCongestion(upd.timestamp(), GetCurrentTimestamp() - pre_wallts);
  }

  void onTradeFeed(const FeedUpdate& upd) {
    double pre_wallts = GetCurrentTimestamp();
    for (auto* callback : callbacks_) {
      callback->onTradeFeed(upd);
    }
    CheckCongestion(upd.timestamp(), GetCurrentTimestamp() - pre_wallts);
  }

  void onLiquidationFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onLiquidationFeed(upd);
    }
  }

  void onOpenInterestFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onOpenInterestFeed(upd);
    }
  }

  void onIndexFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onIndexFeed(upd);
    }
  }

  void onMarkPriceFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onMarkPriceFeed(upd);
    }
  }

  void onFundingRateFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onFundingRateFeed(upd);
    }
  }

  void onIvGreeksFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onIvGreeksFeed(upd);
    }
  }

  void onNavFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onNavFeed(upd);
    }
  }

  void onKlineFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onKlineFeed(upd);
    }
  }

  void onWalletFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onWalletFeed(upd);
    }
  }

  void onMarketAnalysisFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onMarketAnalysisFeed(upd);
    }
  }

  void onTopLongShortAccountRatioFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onTopLongShortAccountRatioFeed(upd);
    }
  }

  void onTopLongShortPositionRatioFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onTopLongShortPositionRatioFeed(upd);
    }
  }

  void onGlobalLongShortAccountRatioFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onGlobalLongShortAccountRatioFeed(upd);
    }
  }

  void onTakerBuySellRatioFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onTakerBuySellRatioFeed(upd);
    }
  }

  void onOpenInterestHistFeed(const FeedUpdate& upd) {
    for (auto* callback : callbacks_) {
      callback->onOpenInterestHistFeed(upd);
    }
  }
 protected:
  std::vector<IFeedSubscriber*> callbacks_;
  typedef presto::math::TimeWindowMovingAverage<int64_t> twmai;
  const int64_t kCongestionWindowNs = 150'000'000'000L;
  const double kCongestionThreshold = 0.35;
  twmai congested_window_ = twmai(86'400'000'000'000L);
  twmai elapsed_tss_ = twmai(kCongestionWindowNs);
  std::optional<int64_t> first_feed_ts_;
  double cum_elapsed_ns_ = 0;
};
}  // namespace coin2::exchange::base::feed
